// Copyright 2020 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// ------------------------------------------------------------------

#include <shared_mem/SharedMemTransport.h>

#include <ertps/utils/LocatorAlias.h>
#include <transport/ReceiverResource.h>
#include <transport/SenderResource.h>
#include <transport/TransportInterface.h>

#include <ertps/rtps/messages/CDRMessage.h>
#include <ertps/rtps/messages/MessageReceiver.h>

#ifdef ANDROID
#include <boostconfig.hpp>
#define ANDROID_UNISTD
#endif  // ifdef ANDROID

#include <utility>
#include <cstring>
#include <algorithm>
#include <memory>

#if defined(__linux) || defined(ANDROID_UNISTD)
#include <unistd.h>
#endif

#if defined(__linux)
#include <sys/stat.h>
#include <dirent.h>
#include <sys/file.h>
#include <string>
#include <cerrno>
#define perror
#endif

#include <elog/Log.hpp>
#include <shared_mem/SHMLocator.hpp>
#include <shared_mem/SharedMemSenderResource.hpp>
#include <shared_mem/SharedMemChannelResource.hpp>
#include <shared_mem/SharedMemManager.hpp>

#define SHM_MANAGER_DOMAIN ("ertps")

namespace evbs {
namespace edds {
namespace rtps {

using octet = ertps::rtps::octet;
using SenderResource = vbs::transport::SenderResource;
using LocatorSelectorEntry = ertps::rtps::LocatorSelectorEntry;
using LocatorSelector = ertps::rtps::LocatorSelector;
using PortParameters = vbsutil::xmlparser::PortParameters;

TransportInterface* SharedMemTransportDescriptor::create_transport(TransportType type) const {
    if (type != TRANSPORT_TYPE_SHM) {
        return nullptr;
    }
    return new SharedMemTransport(*this);
}

//*********************************************************
// SharedMemTransport
//*********************************************************

SharedMemTransport::SharedMemTransport(const SharedMemTransportDescriptor& descriptor)
    : TransportInterface(LOCATOR_KIND_SHM), configuration_(descriptor) {}

SharedMemTransport::SharedMemTransport() : TransportInterface(LOCATOR_KIND_SHM) {}

SharedMemTransport::~SharedMemTransport() {
    // Safely clean already opened resources
    clean_up();
}

bool SharedMemTransport::getDefaultMetatrafficMulticastLocators(LocatorList& locators,
                                                                uint32_t metatraffic_multicast_port) const {
    locators.push_back(SHMLocator::create_locator(metatraffic_multicast_port, SHMLocator::Type::MULTICAST));

    return true;
}

bool SharedMemTransport::getDefaultMetatrafficUnicastLocators(LocatorList& locators,
                                                              uint32_t metatraffic_unicast_port) const {
    locators.push_back(SHMLocator::create_locator(metatraffic_unicast_port, SHMLocator::Type::UNICAST));

    return true;
}

bool SharedMemTransport::getDefaultUnicastLocators(LocatorList& locators, uint32_t unicast_port) const {
    auto locator = SHMLocator::create_locator(unicast_port, SHMLocator::Type::UNICAST);

    (void)fillUnicastLocator(locator, unicast_port);
    locators.push_back(locator);

    return true;
}

void SharedMemTransport::AddDefaultOutputLocator(LocatorList& defaultList) {
    (void)defaultList;
}

const SharedMemTransportDescriptor* SharedMemTransport::configuration() const {
    return &configuration_;
}

bool SharedMemTransport::OpenInputChannel(const Locator& locator, ReceiverResource* receiver, uint32_t maxMsgSize) {
    std::unique_lock<std::recursive_mutex> scopedLock(input_channels_mutex_);

    if (!IsLocatorSupported(locator)) {
        return false;
    }

    if (!IsInputChannelOpen(locator)) {
        try {
            auto channel_resource = CreateInputChannelResource(locator, maxMsgSize, receiver);
            input_channels_.push_back(channel_resource);
        } catch (const boost::interprocess::interprocess_exception& e) {
            (void)e;

            logDebug(RTPS_MSG_OUT, std::string("CreateInputChannelResource failed for port ")
                                       << locator.port << " msg: " << e.what());
            return false;
        } catch (std::exception& e) {
            (void)e;

            logDebug(RTPS_MSG_OUT, std::string("CreateInputChannelResource failed for port ")
                                       << locator.port << " msg: " << e.what());
            return false;
        }
    }

    return true;
}

bool SharedMemTransport::is_locator_allowed(const Locator& locator) const {
    return IsLocatorSupported(locator);
}

LocatorList SharedMemTransport::NormalizeLocator(const Locator& locator) {
    LocatorList list;

    list.push_back(locator);

    return list;
}

bool SharedMemTransport::is_local_locator(const Locator& locator) const {
    assert(locator.kind == LOCATOR_KIND_SHM);
    (void)locator;

    return true;
}

void SharedMemTransport::delete_input_channel(SharedMemChannelResource* channel) {
    channel->disable();
    channel->release();
    channel->clear();
    delete channel;
}

bool SharedMemTransport::CloseInputChannel(const Locator& locator) {
    std::lock_guard<std::recursive_mutex> lock(input_channels_mutex_);

    for (auto it = input_channels_.begin(); it != input_channels_.end(); it++) {
        if ((*it)->locator() == locator) {
            delete_input_channel((*it));
            (void)input_channels_.erase(it);

            return true;
        }
    }

    return false;
}

void SharedMemTransport::clean_up() {
    try {
        // Delete send ports
        opened_ports_.clear();

        // Delete input channels
        {
            std::lock_guard<std::recursive_mutex> lock(input_channels_mutex_);

            for (auto input_channel : input_channels_) {
                delete_input_channel(input_channel);
            }

            input_channels_.clear();
        }

        shared_mem_segment_.reset();
    } catch (const std::exception& e) {
        logWarning(RTPS_MSG_OUT, e.what());
    }
}

bool SharedMemTransport::DoInputLocatorsMatch(const Locator& left, const Locator& right) const {
    return (left.kind == right.kind) && (left.port == right.port);
}

#if defined(__linux)
int SharedMemTransport::shared_mem_segment_clean(const std::string& path, const std::string& domain_name) {
    DIR* dir;
    struct dirent* entry;
    if ((dir = opendir(path.c_str())) == nullptr) {
        logWarning(RTPS_MSG_OUT, "opendir error");
        return -1;
    }
    while ((entry = readdir(dir)) != nullptr) {
        std::string str(entry->d_name);
        if (str.find(domain_name) == std::string::npos) {
            continue;
        }
        /* like ertps_d12489e33825b28b */
        if ((str.find("_port") == std::string::npos) && (str.find("_el") == std::string::npos) &&
            (str.find("_sl") == std::string::npos)) {
            auto lock_file_name_ = str + "_el";
            if (!RobustExclusiveLock::is_locked(lock_file_name_)) {
                // The segment is not locked so the origin is no longer active
                logDebug(RTPS_MSG_OUT, "remove:" << str << " & " << lock_file_name_);
                SharedMemSegment::remove(str.c_str());
            }
            continue;
        }
        /* like ertps_port74121 */
        if ((str.find("_port") != std::string::npos) && (str.find("sem.ertps") == std::string::npos) &&
            (str.find("_el") == std::string::npos) && (str.find("_sl") == std::string::npos)) {
            std::size_t start = str.find_first_of("0123456789");
            std::string port_str = str.substr(start);
            std::size_t el_pos = port_str.find("_el");
            if (el_pos != std::string::npos) {
                (void)port_str.erase(el_pos);
            }
            std::size_t sl_pos = port_str.find("_sl");
            if (sl_pos != std::string::npos) {
                (void)port_str.erase(sl_pos);
            }
            uint32_t port_id = std::stoul(port_str);
            auto port_segment_name = domain_name + "_port" + std::to_string(port_id);
            auto port_lock_name = port_segment_name + "_el";
            auto port_lock_name_sl = port_segment_name + "_sl";
            auto port_mutex_name = "sem." + port_segment_name + "_mutex";
            try {
                deleted_unique_ptr<SharedMemSegment::named_mutex> port_mutex =
                    SharedMemSegment::open_or_create_and_lock_named_mutex(port_segment_name + "_mutex");
                std::unique_lock<SharedMemSegment::named_mutex> port_lock(*port_mutex, std::adopt_lock);
                if (SharedMemGlobal::Port::is_zombie(port_id, domain_name)) {
                    logDebug(RTPS_MSG_OUT, "remove:" << port_segment_name << " & "
                                                     << port_lock_name + "/" + port_lock_name_sl << " & "
                                                     << port_mutex_name);
                    SharedMemSegment::remove(port_segment_name.c_str());
                    (void)SharedMemSegment::named_mutex::remove((port_segment_name + "_mutex").c_str());
                    continue;
                }
            } catch (std::exception& e) {
                (void)e;
                logDebug(RTPS_MSG_OUT, "port:" << port_id << " msg: " << e.what());
            }
            try {
                Locator locator;
                locator.port = port_id;
                auto open_mode = SharedMemGlobal::Port::OpenMode::ReadExclusive;
                auto port = shared_mem_manager_->open_port(locator.port, configuration_.port_queue_capacity(),
                                                           configuration_.healthy_check_timeout_ms(), open_mode);
                auto channel_resource = new SharedMemChannelResource(port->create_listener(), locator, nullptr,
                                                                     configuration_.rtps_dump_file(), false);
                if (channel_resource) {
                    port->set_ref(1);
                    delete channel_resource;
                }
            } catch (std::exception&) {
                logDebug(RTPS_MSG_OUT, "port is used:" << port_id);
            }
        }
    }

    (void)closedir(dir);
    return 0;
}

void SharedMemTransport::shared_mem_clean() {
    std::string path;
    SharedDir::get_default_shared_dir(path);
    (void)shared_mem_segment_clean(path, SHM_MANAGER_DOMAIN);
}
#endif

bool SharedMemTransport::init() {
    // TODO(Adolfo): Calculate this value from UDP sockets buffers size.
    static constexpr uint32_t shm_default_segment_size = 512U * 1024U;

    if (configuration_.segment_size() == 0U) {
        configuration_.segment_size(shm_default_segment_size);
    }

    if (configuration_.segment_size() < configuration_.max_message_size()) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "max_message_size cannot be greater than segment_size");
        return false;
    }

#ifdef ANDROID
    if (access(BOOST_INTERPROCESS_SHARED_DIR_PATH, W_OK) != F_OK) {
        logWarning(RTPS_MSG_OUT,
                   "Unable to write on " << BOOST_INTERPROCESS_SHARED_DIR_PATH << ". SHM Transport not enabled");
        return false;
    }
#endif  // ifdef ANDROID

    try {
        shared_mem_manager_ = SharedMemManager::create(SHM_MANAGER_DOMAIN);
        if (!shared_mem_manager_) {
            return false;
        }
        shared_mem_segment_ =
            shared_mem_manager_->create_segment(configuration_.segment_size(), configuration_.port_queue_capacity());

        // Memset the whole segment to zero in order to force physical map of the buffer
        auto buffer = shared_mem_segment_->alloc_buffer(configuration_.segment_size());
        (void)memset(buffer->data(), 0, configuration_.segment_size());
        buffer.reset();

        if (!configuration_.rtps_dump_file().empty()) {
            auto packets_file_consumer =
                std::unique_ptr<SHMPacketFileConsumer>(new SHMPacketFileConsumer(configuration_.rtps_dump_file()));

            packet_logger_ = std::make_shared<PacketsLog<SHMPacketFileConsumer>>();
            packet_logger_->RegisterConsumer(std::move(packets_file_consumer));
        }
    } catch (std::exception& e) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_ERROR, e.what());
        return false;
    }

    return true;
}

bool SharedMemTransport::IsInputChannelOpen(const Locator& locator) const {
    std::lock_guard<std::recursive_mutex> lock(input_channels_mutex_);

    return IsLocatorSupported(locator) &&
           (std::find_if(input_channels_.begin(), input_channels_.end(), [&](const SharedMemChannelResource* resource) {
                return locator == resource->locator();
            }) != input_channels_.end());
}

bool SharedMemTransport::IsLocatorSupported(const Locator& locator) const {
    return locator.kind == transport_kind_;
}

SharedMemChannelResource* SharedMemTransport::CreateInputChannelResource(const Locator& locator, uint32_t maxMsgSize,
                                                                         ReceiverResource* receiver) {
    (void)maxMsgSize;

    // Multicast locators implies ReadShared (Multiple readers) ports.
    auto open_mode = locator.address[0] == 'M' ? SharedMemGlobal::Port::OpenMode::ReadShared
                                               : SharedMemGlobal::Port::OpenMode::ReadExclusive;

    return new SharedMemChannelResource(shared_mem_manager_
                                            ->open_port(locator.port, configuration_.port_queue_capacity(),
                                                        configuration_.healthy_check_timeout_ms(), open_mode)
                                            ->create_listener(),
                                        locator, receiver, configuration_.rtps_dump_file());
}

bool SharedMemTransport::OpenOutputChannel(SendResourceList& sender_resource_list, const Locator& locator) {
    if (!IsLocatorSupported(locator)) {
        return false;
    }

    // We try to find a SenderResource that can be reuse to this locator.
    // Note: This is done in this level because if we do in NetworkFactory level, we have to mantain
    // what transport already reuses a SenderResource.
    for (auto& sender_resource : sender_resource_list) {
        SharedMemSenderResource* sm_sender_resource = SharedMemSenderResource::cast(*this, sender_resource.get());

        if (sm_sender_resource) {
            return true;
        }
    }

    try {
        sender_resource_list.emplace_back(static_cast<SenderResource*>(new SharedMemSenderResource(*this)));
    } catch (std::exception& e) {
        elogError(
            RTPS_MSG_OUT, RetCode_t::RETCODE_ERROR,
            "SharedMemTransport error opening port " << std::to_string(locator.port) << " with msg: " << e.what());

        return false;
    }

    return true;
}

Locator SharedMemTransport::RemoteToMainLocal(const Locator& remote) const {
    if (!IsLocatorSupported(remote)) {
        return false;
    }

    Locator mainLocal(remote);
    mainLocal.set_Invalid_Address();
    return mainLocal;
}

bool SharedMemTransport::transform_remote_locator(const Locator& remote_locator, Locator& result_locator) const {
    if (IsLocatorSupported(remote_locator)) {
        result_locator = remote_locator;

        return true;
    }

    return false;
}

std::shared_ptr<SharedMemManager::Buffer> SharedMemTransport::copy_to_shared_buffer(const octet* send_buffer,
                                                                                    uint32_t send_buffer_size) {
    assert(shared_mem_segment_);

    std::shared_ptr<SharedMemManager::Buffer> shared_buffer = shared_mem_segment_->alloc_buffer(send_buffer_size);

    (void)memcpy(shared_buffer->data(), send_buffer, send_buffer_size);

    return shared_buffer;
}

bool SharedMemTransport::send(octet* send_buffer, uint32_t send_buffer_size,
                              ertps::rtps::LocatorsIterator* destination_locators_begin,
                              ertps::rtps::LocatorsIterator* destination_locators_end) {
#if !defined(_WIN32)
    cleanup_output_ports();
#endif  // if !defined(_WIN32)

    ertps::rtps::LocatorsIterator& it = *destination_locators_begin;

    bool ret = true;

    std::shared_ptr<SharedMemManager::Buffer> shared_buffer;

    try {
        while (it != *destination_locators_end) {
            if (IsLocatorSupported(*it)) {
                // Only copy the first time
                if (shared_buffer == nullptr) {
                    shared_buffer = copy_to_shared_buffer(send_buffer, send_buffer_size);
                }

                ret &= send(shared_buffer, *it);

                if (packet_logger_ && ret) {
                    packet_logger_->QueueLog({packet_logger_->now(), Locator(), *it, shared_buffer});
                }
            }

            ++it;
        }
    } catch (const std::exception& e) {
        logDebug(RTPS_TRANSPORT_SHM, e.what());
        (void)e;

        // Segment overflow with discard policy doesn't return error.
        if (!shared_buffer) {
            ret = true;
        } else {
            ret = false;
        }
    }

    return ret;
}

void SharedMemTransport::cleanup_output_ports() {
    auto it = opened_ports_.begin();
    while (it != opened_ports_.end()) {
        if (it->second->has_listeners()) {
            ++it;
        } else {
            it = opened_ports_.erase(it);
        }
    }
}

std::shared_ptr<SharedMemManager::Port> SharedMemTransport::find_port(uint32_t port_id) {
    auto ports_it = opened_ports_.find(port_id);

    // The port is already opened
    if (ports_it != opened_ports_.end()) {
        return (*ports_it).second;
    }

    // The port is not opened
    std::shared_ptr<SharedMemManager::Port> port = shared_mem_manager_->open_port(
        port_id, configuration_.port_queue_capacity(), configuration_.healthy_check_timeout_ms(),
        SharedMemGlobal::Port::OpenMode::Write);

    opened_ports_[port_id] = port;

    return port;
}

bool SharedMemTransport::push_discard(const std::shared_ptr<SharedMemManager::Buffer>& buffer,
                                      const Locator& remote_locator) {
    try {
        bool is_port_ok = false;
        const size_t num_retries = 2U;
        for (size_t i = 0U; (i < num_retries) && (!is_port_ok); ++i) {
            if (!find_port(remote_locator.port)->try_push(buffer, is_port_ok)) {
                if (is_port_ok) {
                    logDebug(RTPS_MSG_OUT, "Port " << remote_locator.port << " full. Buffer dropped");
                } else {
                    logWarning(RTPS_MSG_OUT, "Port " << remote_locator.port << " inconsistent. Port dropped");
                    (void)opened_ports_.erase(remote_locator.port);
                }
            }
        }
    } catch (const std::exception& error) {
        logWarning(RTPS_MSG_OUT, error.what());
        return false;
    }

    return true;
}

bool SharedMemTransport::send(const std::shared_ptr<SharedMemManager::Buffer>& buffer, const Locator& remote_locator) {
    if (!push_discard(buffer, remote_locator)) {
        return false;
    }

    logDebug(RTPS_MSG_OUT, "(ID:" << std::this_thread::get_id() << ") "
                                  << "SharedMemTransport: " << buffer->size() << " bytes to port "
                                  << remote_locator.port);

    return true;
}

bool SharedMemTransport::fillMetatrafficMulticastLocator(Locator& locator, uint32_t metatraffic_multicast_port) const {
    if (locator.port == 0U) {
        locator.port = metatraffic_multicast_port;
    }

    return true;
}

bool SharedMemTransport::fillMetatrafficUnicastLocator(Locator& locator, uint32_t metatraffic_unicast_port) const {
    if (locator.port == 0U) {
        locator.port = metatraffic_unicast_port;
    }
    return true;
}

bool SharedMemTransport::configureInitialPeerLocator(Locator& locator, const PortParameters& port_params,
                                                     uint32_t domainId, LocatorList& list) const {
    if (locator.port == 0U) {
        for (uint32_t i = 0U; i < configuration()->maxInitialPeersRange; ++i) {
            Locator auxloc(locator);
            auxloc.port = port_params.getUnicastPort(domainId, i);

            list.push_back(auxloc);
        }
    } else {
        list.push_back(locator);
    }

    return true;
}

bool SharedMemTransport::fillUnicastLocator(Locator& locator, uint32_t well_known_port) const {
    if (locator.port == 0U) {
        locator.port = well_known_port;
    }

    return true;
}

}  // namespace rtps
}  // namespace edds
}  // namespace evbs
